package a_o2odata_deal.utils.zhibo_qy

import a_aa_amainpackage.a_o2odata_deal.config.config.{months, years}
import org.apache.spark.sql.SQLContext

/**
  * @ Auther: o2o-rd-0008
  * @ Date:   2020/4/9 13:55
  * @ Param:  ${PARAM}
  * @ Description: 
  */
object handle_anchor_shop {
  def handleAnchorShop(sqlContext:SQLContext):Unit={
    //加载主播数据
    sqlContext.read.orc(s"s3a://o2o-dataproces-group/panzonghao/tmall/${years}/${months}/all_anchor")
        .selectExpr("anchorId").dropDuplicates().createOrReplaceTempView("anchor")
    //加载原始数据
    sqlContext.read.orc(s"s3a://o2o-dataproces-group/panzonghao/tmall/${years}/${months}/zhibo_finally")
      .selectExpr("platformId","platformName","shopId","shopName","shopUrl","userId").dropDuplicates("userId")
      .createOrReplaceTempView("tmall_data")
    sqlContext.read.orc(s"s3a://o2o-dataproces-group/panzonghao/taobao/${years}/${months}/zhibo_finally")
      .selectExpr("platformId","platformName","shopId","shopName","shopUrl","userId").dropDuplicates("userId")
      .createOrReplaceTempView("taobao_data")

    //计算天猫店铺信息
    val data1 = sqlContext.sql(
      s"""
         |select
         |concat(concat(t2.anchorId,t1.shopId),t1.platformId) as major_key,
         |t2.anchorId,
         |t1.shopId,
         |t1.shopName,
         |t1.shopUrl,
         |t1.platformId,
         |t1.platformName
         |from tmall_data t1
         |inner join
         |anchor t2
         |on t1.userId=t2.anchorId
       """.stripMargin)

    //计算淘宝店铺信息
    val data2 = sqlContext.sql(
      s"""
         |select
         |concat(concat(t2.anchorId,t1.shopId),t1.platformId) as major_key,
         |t2.anchorId,
         |t1.shopId,
         |t1.shopName,
         |t1.shopUrl,
         |t1.platformId,
         |t1.platformName
         |from taobao_data t1
         |inner join
         |anchor t2
         |on t1.userId=t2.anchorId
       """.stripMargin)

    data1.selectExpr("major_key","anchorId","shopId","shopName","shopUrl","platformId","platformName").union(
      data2.selectExpr("major_key","anchorId","shopId","shopName","shopUrl","platformId","platformName")
    ).createOrReplaceTempView("mid")
    sqlContext.read.json(s"s3a://o2o-dataproces-group/panzonghao/tmall/${years}/${months}/lastmonth_anchorshop")
      .createOrReplaceTempView("anchor_shop")
    sqlContext.sql(
      s"""
         |select
         |t1.*
         |from mid t1
         |left join
         |anchor_shop t2
         |on t1.major_key=t2.major_key
         |where t2.major_key is null
       """.stripMargin)
      .repartition(4).write.orc(s"s3a://o2o-dataproces-group/panzonghao/tmall/${years}/${months}/anchor_shop")

  }

}
